package com.aerozhonghuan.archive.service.impl;

import com.aerozhonghuan.archive.config.DynamicDataSourceHolder;
import com.aerozhonghuan.archive.entity.MigrationConfig;
import com.aerozhonghuan.archive.mapper.ArchiveMapper;
import com.aerozhonghuan.archive.mapper.ArchiveMigrationConfigMapper;
import com.aerozhonghuan.archive.mapper.SourceMapper;
import com.aerozhonghuan.archive.mapper.SourceTableMetaMapper;
import com.aerozhonghuan.archive.service.IncrementalMigrationService;
import com.aerozhonghuan.archive.service.MailService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

@Service
public class IncrementalMigrationServiceImpl implements IncrementalMigrationService {

    private static final Logger logger = LoggerFactory.getLogger(IncrementalMigrationServiceImpl.class);

    private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    private final ArchiveMigrationConfigMapper migrationConfigMapper;
    private final SourceMapper sourceMapper;
    private final ArchiveMapper archiveMapper;
    private final SourceTableMetaMapper tableMetaMapper;
    private final MailService mailService;

    public IncrementalMigrationServiceImpl(ArchiveMigrationConfigMapper migrationConfigMapper,
                                           SourceMapper sourceMapper,
                                           ArchiveMapper archiveMapper,
                                           MailService mailService,
                                           SourceTableMetaMapper tableMetaMapper) {
        this.migrationConfigMapper = migrationConfigMapper;
        this.sourceMapper = sourceMapper;
        this.archiveMapper = archiveMapper;
        this.tableMetaMapper = tableMetaMapper;
        this.mailService = mailService;
    }

    /**
     * 增量迁移
     *
     * @param batchSize 每批次处理记录数
     * @return 汇总信息字符串
     */
    @Override
    public String executeIncrementalMigration(int batchSize) {
        StringBuilder summary = new StringBuilder();
        List<MigrationConfig> configs = migrationConfigMapper.getAllEnabled();
        if (configs == null || configs.isEmpty()) {
            logger.warn("未查询到启用的迁移配置，跳过增量迁移。");
            return "未查询到启用的迁移配置，跳过增量迁移。";
        }

        for (MigrationConfig config : configs) {
            // 记录每条配置的开始时间
            Date configStart = new Date();
            logger.info("开始迁移时间：" + sdf.format(configStart));
            summary.append("配置：")
                    .append(config.getSourceDbName()).append(".").append(config.getSourceTableName())
                    .append(" → ")
                    .append(config.getTargetDbName()).append(".").append(config.getTargetTableName())
                    .append("\n开始迁移时间：").append(sdf.format(configStart))
                    .append("\n");

            logger.info("开始处理配置：{}", config);

            try {
                // 初始游标，用于校验和删除
                Date initialCursor = config.getLastCreateTime();
                if (initialCursor == null) {
                    String msg = "last_create_time 为空，跳过该配置。";
                    logger.warn(msg);
                    summary.append(msg).append("\n\n");
                    continue;
                }

                Date cursor = initialCursor;
                Date targetTime = calculateTargetTime();

                // 取列名
                DynamicDataSourceHolder.setDynamicDataSourceKey(config.getSourceDatasource());
                List<String> columnList = tableMetaMapper.getAllColumnNames(
                        config.getSourceDbName(), config.getSourceTableName());
                if (columnList == null || columnList.isEmpty()) {
                    String msg = "未获取到表字段列表，跳过该配置。";
                    logger.warn(msg);
                    summary.append(msg).append("\n\n");
                    continue;
                }
                String columns = String.join(", ", columnList);

                // 游标式分页迁移
                int totalMigrated = 0;
                while (true) {
                    // 从源库查询 batchSize 条记录
                    DynamicDataSourceHolder.setDynamicDataSourceKey(config.getSourceDatasource());
                    List<Map<String, Object>> batch = sourceMapper.fetchByTimeRangeBatch(
                            config.getSourceDbName(),
                            config.getSourceTableName(),
                            config.getTimeStamp(),
                            cursor,
                            targetTime,
                            batchSize
                    );
                    if (batch == null || batch.isEmpty()) {
                        break;
                    }

                    // 构造插入参数
                    List<Object[]> params = new ArrayList<>(batch.size());
                    for (Map<String, Object> row : batch) {
                        Object[] vals = new Object[columnList.size()];
                        for (int i = 0; i < columnList.size(); i++) {
                            vals[i] = row.get(columnList.get(i));
                        }
                        params.add(vals);
                    }

                    // 切换到归档库，批量插入并更新游标
                    DynamicDataSourceHolder.setDynamicDataSourceKey(config.getTargetDatasource());
                    int inserted = archiveMapper.insertBatchGeneric(
                            params,
                            config.getTargetDbName(),
                            config.getTargetTableName(),
                            columns
                    );

                    // 兼容不同时间类型
                    Date newCursor = batch.stream()
                            .map(row -> {
                                Object ts = row.get(config.getTimeStamp());
                                if (ts instanceof Timestamp) {
                                    return (Timestamp) ts;
                                } else if (ts instanceof LocalDateTime) {
                                    return Timestamp.valueOf((LocalDateTime) ts);
                                } else if (ts instanceof Date) {
                                    return new Timestamp(((Date) ts).getTime());
                                } else {
                                    throw new IllegalArgumentException("Unsupported time type: " + ts.getClass());
                                }
                            })
                            .max(Timestamp::compareTo)
                            .map(ts -> new Date(ts.getTime()))
                            .orElseThrow(() -> new IllegalStateException("无法获取最大时间戳"));

                    DynamicDataSourceHolder.setDynamicDataSourceKey(config.getTargetDatasource());
                    migrationConfigMapper.updateLastCreateTimeByNewTime(config.getId(), newCursor);

                    totalMigrated += inserted;
                    cursor = newCursor;
                    logger.info("已迁移 {} 条，游标更新至 {}", totalMigrated, sdf.format(cursor));
                }

                // 记录配置结束时间
                Date configEnd = new Date();
                summary.append("结束迁移时间：").append(sdf.format(configEnd)).append("\n\n");
                logger.info("结束迁移时间：" + sdf.format(configEnd));
                summary.append("本次迁移数量：").append(totalMigrated).append("\n\n");

                // ---- 迁移完成，执行校验 ----
                long sourceCount = getSourceCount(
                        config.getSourceDbName(),
                        config.getSourceTableName(),
                        config.getTimeStamp(),
                        initialCursor,
                        targetTime,
                        config.getSourceDatasource()
                );
                long archiveCount = getArchiveCount(
                        config.getTargetDbName(),
                        config.getTargetTableName(),
                        config.getTimeStamp(),
                        initialCursor,
                        targetTime,
                        config.getTargetDatasource()
                );

                // 记录校验结果
                if (sourceCount == archiveCount) {
                    summary.append(String.format(
                            "校验通过：源库=%d，归档库=%d\n",
                            sourceCount, archiveCount));

                    // 校验通过后删除源库已迁移数据
                    DynamicDataSourceHolder.setDynamicDataSourceKey(config.getSourceDatasource());
                    List<Long> idList = sourceMapper.fetchIdsByTimeRange(
                            config.getSourceDbName(),
                            config.getSourceTableName(),
                            config.getTimeStamp(),
                            initialCursor,
                            targetTime
                    );
                    int totalDeleted = 0;
                    if (idList != null && !idList.isEmpty()) {
                        for (int i = 0; i < idList.size(); i += batchSize) {
                            int end = Math.min(i + batchSize, idList.size());
                            List<Long> sub = idList.subList(i, end);
                            DynamicDataSourceHolder.setDynamicDataSourceKey(config.getSourceDatasource());
                            int deleted = sourceMapper.deleteSourceRecordsByMigration(
                                    config.getSourceDbName(),
                                    config.getSourceTableName(),
                                    config.getTimeStamp(),
                                    initialCursor,
                                    targetTime,
                                    sub
                            );
                            totalDeleted += deleted;
                        }
                    }
                    summary.append(String.format("已删除源库 %d 条已迁移数据\n", totalDeleted));

                } else {
                    summary.append(String.format(
                            "校验失败，不执行删除操作：源库=%d，归档库=%d\n",
                            sourceCount, archiveCount));
                }

            } catch (Exception e) {
                String err = String.format(
                        "配置 %s.%s → %s.%s 迁移失败：%s\n\n",
                        config.getSourceDbName(), config.getSourceTableName(),
                        config.getTargetDbName(), config.getTargetTableName(),
                        e.getMessage()
                );
                logger.error(err, e);
                summary.append(err);
                try {
                    mailService.sendEmail("增量迁移失败通知", err);
                } catch (Exception mailEx) {
                    logger.error("失败通知邮件发送失败", mailEx);
                }
            }
        }

        String result = summary.toString();
        try {
            mailService.sendEmail("增量数据迁移完成", result);
        } catch (Exception e) {
            logger.error("完成通知邮件发送失败", e);
        }
        logger.info("增量迁移汇总结果：\n{}", result);
        return result;
    }

    /**
     * 计算“前一年同季度1号零点”时间
     */
    private Date calculateTargetTime() {
        LocalDate today = LocalDate.now();
        int qStartMonth = ((today.getMonthValue() - 1) / 3) * 3 + 1;
        LocalDate targetDate = LocalDate.of(today.getYear() - 1, qStartMonth, 1);
        return Date.from(targetDate.atStartOfDay(ZoneId.systemDefault()).toInstant());
    }


    /**
     * 校验源库记录数
     */
    private long getSourceCount(String schema, String table, String tsColumn,
                                Date start, Date end, String dsKey) {
        DynamicDataSourceHolder.setDynamicDataSourceKey(dsKey);
        return sourceMapper.countByMigration(schema, table, tsColumn, start, end);
    }

    /**
     * 校验归档库记录数
     */
    private long getArchiveCount(String schema, String table, String tsColumn,
                                 Date start, Date end, String dsKey) {
        DynamicDataSourceHolder.setDynamicDataSourceKey(dsKey);
        return archiveMapper.countByMigration(schema, table, tsColumn, start, end);
    }
}
